package com.aotain.nyx.abnormal;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.log4j.Logger;

public class AbnormalReduce implements ReduceFunction<Tuple2<String,AbnormalStatTuple>>{

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = -6119926283638667302L;
	
	/**
	 * reduce执行过程:v1,v2,v3,v4.....
	 * 				  v1+v2(v1_),v3(v2_),v4(v3_)...
	 * 				  v1_+v2_(v1__)<v1+v2+v3>,v4(v2__)...
	 * 				  ...
	 */	
	@Override
	public Tuple2<String, AbnormalStatTuple> reduce(
			Tuple2<String, AbnormalStatTuple> value1,
			Tuple2<String, AbnormalStatTuple> value2) throws Exception {
		if(value1 == null || value2 == null)
			return null; 
		
		HashMap<String,Long> dipPVMap = value1.f1.getDipPVMap()==null?new HashMap<String,Long>():value1.f1.getDipPVMap();//目标IP PV数
		HashMap<String,Long> sdPVMap = value1.f1.getSdPVMap()==null?new HashMap<String,Long>():value1.f1.getSdPVMap();//各个源ip到目标ip的PV数
		Map<String,Long> tripleMap = value1.f1.getTripleMap()==null?new HashMap<String,Long>():value1.f1.getTripleMap();//各个元组的PV
		Map<String,Long> maxDelayMap = value1.f1.getMaxDelayMap()==null?new HashMap<String,Long>():value1.f1.getMaxDelayMap();//最大延时Map
		Map<String,Long> minDelayMap = value1.f1.getMinDelayMap()==null?new HashMap<String,Long>():value1.f1.getMinDelayMap();//最小延时Map
		Map<String,Long> streamPacketMap = value1.f1.getStreamPacketMap()==null?new HashMap<String,Long>():value1.f1.getStreamPacketMap();//
		Map<String,Long> streamOctetsMap = value1.f1.getStreamOctetsMap()==null?new HashMap<String,Long>():value1.f1.getStreamOctetsMap();//
		Map<String,Integer> delayTimePVMap = value1.f1.getDelayTimePVMap()==null?new HashMap<String,Integer>():value1.f1.getDelayTimePVMap();//		
		Map<String,Long> delayTimeMap = value1.f1.getDelaytimeMap()==null?new HashMap<String,Long>():value1.f1.getDelaytimeMap();//求延时Map
		Map<String,Set<String>> dipToTriple = value1.f1.getDipToTriple()==null?new HashMap<String,Set<String>>():value1.f1.getDipToTriple();//目标ip下的三元组映射
		String tripleKey1 = null;
		String tripleKey2 = null;
		String sdipKey1 = null;
		String sdipKey2 = null;
		
		Tuple2<String,AbnormalStatTuple> retTuple = new Tuple2<String,AbnormalStatTuple>();
//		Logger.getLogger(getClass()).info("#####AbnormalStatMap#### tuple is " );
		String idc = value1.f1.getIDC();
		String dip = value1.f1.getDestIP();
		String dport = value1.f1.getDestPort();
		String sip1 = value1.f1.getSourceIP();
		String gis1 = value1.f1.getGIS();
		String type1 = value1.f1.getType();
		String flowDir = value1.f1.getFlowDirection();
		Long delayTime1 = value1.f1.getDelayTime();
		long upStreamOctets1 = value1.f1.getUpStreamOctets();
		long upStreamPacket1 = value1.f1.getUpStreamPacket();
		
		String sip2 = value2.f1.getSourceIP();
		String gis2 = new String(value2.f1.getGIS());
		String type2 = value2.f1.getType();
		Long delayTime2 = value2.f1.getDelayTime();
		long upStreamOctets2 = value2.f1.getUpStreamOctets();
		long upStreamPacket2 = value2.f1.getUpStreamPacket();
		
		String idc_dip_dport = dip+"*"+idc+"*"+dport;
		if(sip1 != null && gis1 != null && type1 != null){
			tripleKey1 = dip + "*" + dport + "*" + sip1 + "*" + gis1 + "*" + type1 + "*" + idc + "*" + flowDir;
			sdipKey1 = dip + "*" + idc + "*" + sip1;
		}
		tripleKey2 = dip + "*" + dport + "*" + sip2 + "*" + gis2 + "*" + type2 + "*" + idc + "*" + flowDir;
		sdipKey2 = dip + "*" + idc + "*" + sip2;
//		streamPacketMap.put(tripleKey, upStreamPacket);
//		streamOctetsMap.put(tripleKey, upStreamOctets);
//		tripleMap.put(tripleKey, upStreamOctets);
		
//		long upStreamOctets = value1.f1.getUpStreamOctets() + value2.f1.getUpStreamOctets();
		long upStreamPacket = value1.f1.getUpStreamPacket() + value2.f1.getUpStreamPacket();
		long upStreamOctets = value1.f1.getUpStreamOctets() + value2.f1.getUpStreamOctets();
		long dipPV = upStreamPacket;
		long triplePV = upStreamOctets1;
		long sdPV = upStreamPacket1;
		long triplePacket = upStreamPacket1;
		long tripleOctets= upStreamOctets1;
		int delayPV = 1;
		
		/**dipPVMap**/
		if(value1.f1.getDipPVMap().containsKey(idc_dip_dport)) {
			dipPV += value1.f1.getDipPVMap().get(idc_dip_dport);
		}
		dipPVMap.put(idc_dip_dport, dipPV);
		
		
		/**tripleMap**/
		if(tripleKey1 != null && !value1.f1.getTripleMap().containsKey(tripleKey1)) {//这里非空判断是判断是否为第一次进入，后面过程中这些属性值为空 因为没赋值
//			if(value1.f1.getTripleMap().containsKey(tripleKey1)) {
//				triplePV += value1.f1.getTripleMap().get(tripleKey1); 
//			}
			tripleMap.put(tripleKey1, triplePV);
		}
		triplePV = upStreamOctets2;
		if(value1.f1.getTripleMap().containsKey(tripleKey2)) {
			triplePV += value1.f1.getTripleMap().get(tripleKey2); 
		}
		tripleMap.put(tripleKey2, triplePV);
		
		
		/**sdPVMap**/
		if(sdipKey1 != null && !value1.f1.getSdPVMap().containsKey(sdipKey1)) {
//			if(value1.f1.getSdPVMap().containsKey(sdipKey1)) {
//				sdPV += value1.f1.getSdPVMap().get(sdipKey1);
//			}
			sdPVMap.put(sdipKey1, sdPV);
		}
		sdPV = upStreamPacket2;
		if(value1.f1.getSdPVMap().containsKey(sdipKey2)) {
			sdPV += value1.f1.getSdPVMap().get(sdipKey2);
		}
		sdPVMap.put(sdipKey2, sdPV);
		
		
		/**streamPacketMap**/
		if(tripleKey1 != null && !value1.f1.getStreamPacketMap().containsKey(tripleKey1)) {
//			if(value1.f1.getStreamPacketMap().containsKey(tripleKey1)){
//				triplePacket += value1.f1.getStreamPacketMap().get(tripleKey1);
//			}
			streamPacketMap.put(tripleKey1, triplePacket);
		}
		triplePacket = upStreamPacket2;
		if(value1.f1.getStreamPacketMap().containsKey(tripleKey2)) {
			triplePacket += value1.f1.getStreamPacketMap().get(tripleKey2);
		}
		streamPacketMap.put(tripleKey2, triplePacket);
		
		
		/**streamOctetsMap**/
		if(tripleKey1 != null && !value1.f1.getStreamOctetsMap().containsKey(tripleKey1)) {
//			if(value1.f1.getStreamOctetsMap().containsKey(tripleKey1)){
//				tripleOctets += value1.f1.getStreamOctetsMap().get(tripleKey1);
//			}
			streamOctetsMap.put(tripleKey1, tripleOctets);
		}
		tripleOctets = upStreamOctets2;
		if(value1.f1.getStreamOctetsMap().containsKey(tripleKey2)) {
			tripleOctets += value1.f1.getStreamOctetsMap().get(tripleKey2);
		}
		streamOctetsMap.put(tripleKey2, tripleOctets);
		
		
		/**delayTimePVMap**/
		if(sdipKey1 != null && !value1.f1.getDelayTimePVMap().containsKey(sdipKey1)) {
//			if(value1.f1.getDelayTimePVMap().containsKey(sdipKey1)) {
//				delayPV = value1.f1.getDelayTimePVMap().get(sdipKey1) + 1;
//			}
			delayTimePVMap.put(sdipKey1, delayPV);
		}
		if(value1.f1.getDelayTimePVMap().containsKey(sdipKey2)) {
			delayPV = value1.f1.getDelayTimePVMap().get(sdipKey2) + 1;
		}
		delayTimePVMap.put(sdipKey2, delayPV);
		
		
		/**delayTimeMap**/
		if(sdipKey1 != null && !value1.f1.getDelaytimeMap().containsKey(sdipKey1)) {
//			if(value1.f1.getDelaytimeMap().containsKey(sdipKey1)) {
//				delayTime1 += value1.f1.getDelaytimeMap().get(sdipKey1);
//			}
			delayTimeMap.put(sdipKey1, delayTime1);
		}
		if(value1.f1.getDelaytimeMap().containsKey(sdipKey2)) {
			delayTime2 += value1.f1.getDelaytimeMap().get(sdipKey2);
		}
		delayTimeMap.put(sdipKey2, delayTime2);
		
		
		/**maxDelayMap**/
		if(sdipKey1 != null && !value1.f1.getMaxDelayMap().containsKey(sdipKey1)) {
//			if(value1.f1.getMaxDelayMap().containsKey(sdipKey1)) {
//				if(delayTime1>value1.f1.getMaxDelayMap().get(sdipKey1)) {
					maxDelayMap.put(sdipKey1, delayTime1);
//				}
//			}
		}
		if(value1.f1.getMaxDelayMap().containsKey(sdipKey2)) {
			if(delayTime2>value1.f1.getMaxDelayMap().get(sdipKey2)) {
				maxDelayMap.put(sdipKey2, delayTime2);
			}
		}else {
			maxDelayMap.put(sdipKey2, delayTime2);
		}
		
		
		/**minDelayMap**/
		if(sdipKey1 != null && !value1.f1.getMinDelayMap().containsKey(sdipKey1)) {
//			if(value1.f1.getMinDelayMap().containsKey(sdipKey1)) {
//				if(delayTime1<value1.f1.getMinDelayMap().get(sdipKey1)) {
					minDelayMap.put(sdipKey1, delayTime1);
//				}
//			}
		}
		if(value1.f1.getMinDelayMap().containsKey(sdipKey2)) {
			if(delayTime2<value1.f1.getMinDelayMap().get(sdipKey2)) {
				minDelayMap.put(sdipKey2, delayTime2);
			}
		}else {
			minDelayMap.put(sdipKey2, delayTime2);
		}
		
		
		/**dipToTriple**/
		Set<String> tripleSet = dipToTriple.get(idc_dip_dport);
		if(tripleSet == null) {
			tripleSet = new HashSet<String>();
		}
		if(tripleKey1 != null && !tripleSet.contains(tripleKey1)) {
			tripleSet.add(tripleKey1);
		}
		tripleSet.add(tripleKey2);
		dipToTriple.put(idc_dip_dport, tripleSet);
		
//		if(tripleKey1==null || tripleKey2==null) {
//			Logger.getLogger(getClass()).info("#####AbnormalReduce#### tripleKey1 tuple  " + value1.f1.toString()
//					+ "tripleKey2 tuple" + value2.f1.toString());
//		}

		AbnormalStatTuple stat = new AbnormalStatTuple();
		stat.setIDC(idc);
		stat.setDestIP(dip);
		stat.setDestPort(dport);
		stat.setUpStreamOctets(upStreamOctets);
		stat.setUpStreamPacket(upStreamPacket);
		stat.setStreamOctetsMap(streamOctetsMap);
		stat.setStreamPacketMap(streamPacketMap);
		stat.setDipToTriple(dipToTriple);
		stat.setSdPVMap(sdPVMap);
		stat.setDipPVMap(dipPVMap);
		stat.setTripleMap(tripleMap);
		stat.setDelaytimeMap(delayTimeMap);
		stat.setMaxDelayMap(maxDelayMap);
		stat.setMinDelayMap(minDelayMap);
		stat.setDelayTimePVMap(delayTimePVMap);
		
		retTuple.setFields(idc_dip_dport, stat);
		return retTuple;
	}

}
